package com.shujia.flink.core

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

import java.text.SimpleDateFormat
import java.time.Duration
import java.util.Date

object Demo4EventTIme {
  def main(args: Array[String]): Unit = {


    /*
java,2022-08-29 16:33:10
java,2022-08-29 16:33:11
java,2022-08-29 16:33:12
java,2022-08-29 16:33:13
java,2022-08-29 16:33:15
java,2022-08-29 16:33:14
java,2022-08-29 16:33:20
java,2022-08-29 16:33:16
java,2022-08-29 16:33:17
java,2022-08-29 16:33:18
java,2022-08-29 16:33:19
java,2022-08-29 16:33:22
java,2022-08-29 16:33:27


     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val wordsDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 整理数据
     */

    val wordAndTimeDS: DataStream[(String, String)] = wordsDS.map(line => {
      val split: Array[String] = line.split(",")
      val word: String = split(0)
      val timeStr: String = split(1)
      (word, timeStr)
    })


    //时间字段需要时间戳
    //告诉flink哪一个字段是时间字段，时间字段需要一个毫秒级别的时间戳
    /*val assDS: DataStream[(String, String)] = wordAndTimeDS.assignAscendingTimestamps(kv => {
      val timeStr: String = kv._2
      val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
      //将时间字符串转换成时间对象
      val date: Date = format.parse(timeStr)
      //获取时间戳
      val time: Long = date.getTime
      //返回事件时间
      time
    })*/

    /**
     * 数据真实发送的顺序在经常数据采集和传输之后会乱序，当数据乱序了，flink在计算时可能会丢失数据
     * 解决乱序的方法：将水位线前移
     * 水位线默认等于最新一条数据的时间戳
     * 窗口的触发条件是：水位线大于等于窗口的结束时间，窗口内有数据
     *
     */

    val assDS: DataStream[(String, String)] = wordAndTimeDS
      .assignTimestampsAndWatermarks(WatermarkStrategy
        //设置水位线前移的时间，设置数据最大乱序的时间
        .forBoundedOutOfOrderness[(String, String)](Duration.ofSeconds(5))
        //设置时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, String)] {
          override def extractTimestamp(kv: (String, String), recordTimestamp: Long): Long = {
            val timeStr: String = kv._2
            val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
            //将时间字符串转换成时间对象
            val date: Date = format.parse(timeStr)
            //获取时间戳
            val time: Long = date.getTime
            //返回事件时间
            time
          }
        }))


    //去掉时间转在后面拼接1
    val kvDS: DataStream[(String, Int)] = assDS.map(kv => (kv._1, 1))

    /**
     * 使用事件时间每隔5秒统计单词的数量
     *
     */

    //安装单词分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //TumblingEventTimeWindows: 滚动的事件时间窗口
    val windiwDS: WindowedStream[(String, Int), String, TimeWindow] = keyByDS
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))

    //统计
    val countDS: DataStream[(String, Int)] = windiwDS.sum(1)

    countDS.print()

    env.execute()
  }

}
